package com.datagateway.component;

import com.datagateway.model.ProcessedData;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Elasticsearch数据源实现
 * 
 * @author Data Gateway Team
 * @version 1.0.0
 */
public class ElasticsearchDataSource implements DataSourceManager.DataSourceInstance {

    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchDataSource.class);

    private final Map<String, Object> config;
    private final String type = "ELASTICSEARCH";
    private RestHighLevelClient client;
    private boolean connected = false;

    public ElasticsearchDataSource(Map<String, Object> config) {
        this.config = new ConcurrentHashMap<>(config);
        initialize();
    }

    /**
     * 初始化Elasticsearch连接
     */
    private void initialize() {
        try {
            String host = (String) config.getOrDefault("host", "localhost");
            int port = (Integer) config.getOrDefault("port", 9200);
            String scheme = (String) config.getOrDefault("scheme", "http");
            String username = (String) config.get("username");
            String password = (String) config.get("password");

            // 构建连接
            RestClientBuilder builder = RestClient.builder(
                new org.apache.http.HttpHost(host, port, scheme)
            );

            // 设置认证
            if (username != null && password != null) {
                builder.setHttpClientConfigCallback(httpClientBuilder -> {
                    httpClientBuilder.setDefaultCredentialsProvider(
                        new org.apache.http.impl.client.BasicCredentialsProvider()
                    );
                    return httpClientBuilder;
                });
            }

            client = new RestHighLevelClient(builder);
            connected = true;

            logger.info("Elasticsearch数据源初始化成功: {}://{}:{}", scheme, host, port);

        } catch (Exception e) {
            logger.error("Elasticsearch数据源初始化失败: {}", e.getMessage(), e);
            connected = false;
        }
    }

    @Override
    public boolean testConnection() {
        try {
            if (client == null) {
                initialize();
            }

            if (client != null) {
                // 执行ping操作测试连接
                client.ping(RequestOptions.DEFAULT);
                connected = true;
                return true;
            }
            return false;

        } catch (IOException e) {
            logger.error("Elasticsearch连接测试失败: {}", e.getMessage(), e);
            connected = false;
            return false;
        }
    }

    @Override
    public boolean writeData(List<ProcessedData> dataList) {
        if (!connected || client == null) {
            logger.error("Elasticsearch连接未建立");
            return false;
        }

        if (dataList == null || dataList.isEmpty()) {
            return true;
        }

        try {
            String indexName = (String) config.getOrDefault("index", "datagateway");
            BulkRequest bulkRequest = new BulkRequest();

            for (ProcessedData data : dataList) {
                IndexRequest indexRequest = new IndexRequest(indexName);
                indexRequest.id(data.getId());
                
                // 构建文档
                Map<String, Object> document = new HashMap<>();
                document.put("id", data.getId());
                document.put("content", data.getContent());
                document.put("timestamp", data.getTimestamp());
                document.put("created_time", new Date());
                
                if (data.getMetadata() != null) {
                    document.put("metadata", data.getMetadata());
                }

                indexRequest.source(document, XContentType.JSON);
                bulkRequest.add(indexRequest);
            }

            // 执行批量插入
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            
            if (bulkResponse.hasFailures()) {
                logger.error("Elasticsearch批量插入有失败项: {}", bulkResponse.buildFailureMessage());
                return false;
            }

            logger.info("Elasticsearch数据写入成功: count={}", dataList.size());
            return true;

        } catch (IOException e) {
            logger.error("Elasticsearch数据写入失败: {}", e.getMessage(), e);
            return false;
        }
    }

    @Override
    public List<ProcessedData> readData(String query, Map<String, Object> parameters) {
        if (!connected || client == null) {
            logger.error("Elasticsearch连接未建立");
            return Collections.emptyList();
        }

        List<ProcessedData> resultList = new ArrayList<>();

        try {
            String indexName = (String) config.getOrDefault("index", "datagateway");
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            // 解析查询条件
            if (query != null && !query.trim().isEmpty()) {
                if (query.contains("=")) {
                    // 等值查询
                    String[] parts = query.split("=");
                    if (parts.length == 2) {
                        String field = parts[0].trim();
                        String value = parts[1].trim();
                        
                        // 处理参数替换
                        if (value.startsWith(":")) {
                            String paramName = value.substring(1);
                            Object paramValue = parameters.get(paramName);
                            if (paramValue != null) {
                                searchSourceBuilder.query(QueryBuilders.termQuery(field, paramValue));
                            }
                        } else {
                            searchSourceBuilder.query(QueryBuilders.termQuery(field, value));
                        }
                    }
                } else {
                    // 全文搜索
                    searchSourceBuilder.query(QueryBuilders.matchQuery("content", query));
                }
            } else {
                // 查询所有
                searchSourceBuilder.query(QueryBuilders.matchAllQuery());
            }

            // 设置返回数量
            int size = (Integer) config.getOrDefault("searchSize", 100);
            searchSourceBuilder.size(size);

            searchRequest.source(searchSourceBuilder);

            // 执行搜索
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            
            for (SearchHit hit : searchResponse.getHits().getHits()) {
                ProcessedData data = convertSearchHitToProcessedData(hit);
                resultList.add(data);
            }

            logger.info("Elasticsearch数据读取成功: count={}", resultList.size());
            return resultList;

        } catch (IOException e) {
            logger.error("Elasticsearch数据读取失败: {}", e.getMessage(), e);
            return Collections.emptyList();
        }
    }

    @Override
    public boolean executeSql(String sql, Map<String, Object> parameters) {
        if (!connected || client == null) {
            logger.error("Elasticsearch连接未建立");
            return false;
        }

        try {
            // Elasticsearch不支持传统SQL，这里可以执行一些管理操作
            if (sql.toLowerCase().startsWith("create index")) {
                String indexName = extractIndexName(sql);
                // 创建索引的逻辑
                logger.info("Elasticsearch索引创建: {}", indexName);
                return true;
            } else if (sql.toLowerCase().startsWith("delete index")) {
                String indexName = extractIndexName(sql);
                // 删除索引的逻辑
                logger.info("Elasticsearch索引删除: {}", indexName);
                return true;
            } else {
                logger.warn("Elasticsearch不支持SQL语句: {}", sql);
                return false;
            }

        } catch (Exception e) {
            logger.error("Elasticsearch命令执行失败: sql={}, error={}", sql, e.getMessage(), e);
            return false;
        }
    }

    @Override
    public boolean isConnected() {
        try {
            if (client == null) {
                return false;
            }
            client.ping(RequestOptions.DEFAULT);
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    @Override
    public void close() {
        try {
            if (client != null) {
                client.close();
            }
            connected = false;
            logger.info("Elasticsearch数据源连接已关闭");
        } catch (IOException e) {
            logger.error("关闭Elasticsearch连接失败: {}", e.getMessage(), e);
        }
    }

    @Override
    public String getType() {
        return type;
    }

    @Override
    public Map<String, Object> getConfig() {
        return new HashMap<>(config);
    }

    /**
     * 将SearchHit转换为ProcessedData
     */
    private ProcessedData convertSearchHitToProcessedData(SearchHit hit) {
        ProcessedData data = new ProcessedData();
        Map<String, Object> sourceMap = hit.getSourceAsMap();
        
        data.setId((String) sourceMap.get("id"));
        data.setContent((String) sourceMap.get("content"));
        
        Object timestamp = sourceMap.get("timestamp");
        if (timestamp instanceof Number) {
            data.setTimestamp(((Number) timestamp).longValue());
        }
        
        @SuppressWarnings("unchecked")
        Map<String, Object> metadata = (Map<String, Object>) sourceMap.get("metadata");
        data.setMetadata(metadata);
        
        return data;
    }

    /**
     * 从SQL语句中提取索引名称
     */
    private String extractIndexName(String sql) {
        try {
            String[] parts = sql.split("\\s+");
            for (int i = 0; i < parts.length - 1; i++) {
                if ("index".equalsIgnoreCase(parts[i])) {
                    return parts[i + 1];
                }
            }
            return "datagateway"; // 默认索引名
        } catch (Exception e) {
            logger.error("提取索引名称失败: sql={}, error={}", sql, e.getMessage(), e);
            return "datagateway";
        }
    }
}
